package com.zmn.oms.business.impl.report;

import com.alibaba.fastjson.JSONObject;
import com.zmn.common.utils.validator.ValidatorUtil;
import com.zmn.dms.common.enums.DiscountTypeEnums;
import com.zmn.oms.business.interfaces.es.OrderWorkEsBService;
import com.zmn.oms.business.interfaces.report.CoupBatchStatisticsBService;
import com.zmn.oms.model.entity.report.CoupBatchFact;
import com.zmn.oms.model.entity.report.CoupBatchQuery;
import com.zmn.oms.model.entity.report.CoupUserWork;
import com.zmn.pay.common.constant.PayConsts;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.ListUtils;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.filter.ParsedFilter;
import org.elasticsearch.search.aggregations.bucket.nested.ParsedNested;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedLongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.ParsedSum;
import org.elasticsearch.search.aggregations.metrics.ParsedTopHits;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.data.domain.Pageable;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 优惠劵批次与工单相关度量统计实现
 *
 * @author hankangshuai
 * @since 2019/11/5
 **/
@Slf4j
@Service
public class CoupBatchStatisticsBServiceImpl implements CoupBatchStatisticsBService {
    /**
     * 分区大小
     */
    public static final int PARTITION_SIZE = 1000;

    @Resource
    OrderWorkEsBService orderWorkEsBService;
    /**
     * 所需字段
     */
    private String[] REQUIRED_FIELD;
    public static final String LOST_PARAMETER_MSG = "[优惠劵统计接口]参数缺失";
    /**
     * 单次查询最大数量
     */
    public static final int MAX_COUP_USE = 500;

    public CoupBatchStatisticsBServiceImpl() {
        this.REQUIRED_FIELD = new String[]{"userId", "id"};
    }

    @Override
    public CoupBatchFact findCoupBatchFact(CoupBatchQuery coupBatchQuery) {
        ValidatorUtil.validateWithException(coupBatchQuery);
        Integer batchID = coupBatchQuery.getBatchID();
        List<CoupUserWork> userWorkMaps = coupBatchQuery.getUserWorkMaps();
        try {
            if (userWorkMaps.size() > MAX_COUP_USE) {
                List<List<CoupUserWork>> partition = ListUtils.partition(userWorkMaps, PARTITION_SIZE);
                return partition.stream().map(coupUserWorks -> this.getCoupBatchFact(coupUserWorks, batchID))
                        .reduce(new CoupBatchFact(), (coupBatchFact1, coupBatchFact2) -> {
                            coupBatchFact1.setBatchId(
                                    Optional.ofNullable(coupBatchFact1.getBatchId()).orElse(coupBatchFact2.getBatchId())
                            );
                            coupBatchFact1.setNewCount(
                                    Optional.ofNullable(coupBatchFact1.getNewCount()).orElse(0) +
                                            Optional.ofNullable(coupBatchFact2.getNewCount()).orElse(0)
                            );
                            coupBatchFact1.setRepurchaseCount(
                                    Optional.ofNullable(coupBatchFact1.getRepurchaseCount()).orElse(0) +
                                            Optional.ofNullable(coupBatchFact2.getRepurchaseCount()).orElse(0)
                            );
                            coupBatchFact1.setDiscountAmount(
                                    Optional.ofNullable(coupBatchFact1.getDiscountAmount()).orElse(0L) +
                                            Optional.ofNullable(coupBatchFact2.getDiscountAmount()).orElse(0L)
                            );
                            coupBatchFact1.setOriginalAmount(
                                    Optional.ofNullable(coupBatchFact1.getOriginalAmount()).orElse(0L) +
                                            Optional.ofNullable(coupBatchFact2.getOriginalAmount()).orElse(0L)
                            );
                            return coupBatchFact1;
                        });
            } else {
                return this.getCoupBatchFact(userWorkMaps, batchID);
            }
        } catch (Exception e) {
            log.error("【优惠劵报表统计】预期外的情况，入参：{}", JSONObject.toJSONString(coupBatchQuery));
            throw e;
        }
    }

    public CoupBatchFact getCoupBatchFact(List<CoupUserWork> userWorkMaps, Integer batchID) {
        List<Long> allWorkIds = userWorkMaps.stream().map(CoupUserWork::getWorkId).collect(Collectors.toList());
        //用户 与 用户首次使用了优惠劵的工单映射
        Map<Long, CoupUserWork> userWorkMap = userWorkMaps.stream()
                .collect(Collectors.toMap(CoupUserWork::getUserId, Function.identity(), (o, o2) -> {
                    long timeA = Objects.requireNonNull(o.getUseTime(), LOST_PARAMETER_MSG).getTime();
                    long timeB = Objects.requireNonNull(o2.getUseTime(), LOST_PARAMETER_MSG).getTime();
                    return (timeA <= timeB) ? o : o2;
                }));
        NativeSearchQuery newCountQuery = new NativeSearchQueryBuilder()
                .withFields(REQUIRED_FIELD)
                .withPageable(Pageable.unpaged())
                .withQuery(QueryBuilders.termsQuery("userId", userWorkMap.keySet()))
                .addAggregation(
                        AggregationBuilders.terms("userIdTerm").field("userId").size(10000)
                                .subAggregation(
                                        AggregationBuilders.topHits("firstWork").size(1)
                                                .fetchSource(new String[]{"id"}, new String[0])
                                                .sort(SortBuilders.fieldSort("inputTime").order(SortOrder.ASC))
                                )
                ).addAggregation(
                        AggregationBuilders.filter("amountSumFilter", QueryBuilders.boolQuery()
                                .must(QueryBuilders.termsQuery("id", allWorkIds))
                                .must(QueryBuilders.termQuery("payStatus", PayConsts.PAY_STATUS_DONE))
                        )
                                .subAggregation(
                                        AggregationBuilders.nested("discountList", "discountList")
                                                .subAggregation(
                                                        AggregationBuilders.filter("filterCoupon", QueryBuilders.termQuery("discountList.discountCateg", DiscountTypeEnums.DISCOUNT_COUP_COMMON.getSubType()))
                                                                .subAggregation(AggregationBuilders.sum("discountAmountSum").field("discountList.discountAmount"))
                                                )
                                )
                                .subAggregation(AggregationBuilders.sum("originalAmountSum").field("originalAmount"))
                )
                .build();
        Aggregations aggregations = orderWorkEsBService.getAggregationsByQuery(newCountQuery);
        CoupBatchFact coupBatchFact = new CoupBatchFact();
        coupBatchFact.setBatchId(batchID);
        ParsedTerms parsedTerms = aggregations.get("userIdTerm");
        List<? extends Terms.Bucket> buckets = parsedTerms.getBuckets();

        //noinspection DuplicatedCode
        List<? extends Terms.Bucket> newCountBucket = buckets.stream()
                // 过滤用户首次使用优惠劵的工单ID与该用户首次工单ID不一致的工单
                .filter(bucket -> {
                    Number keyAsNumber = bucket.getKeyAsNumber();
                    ParsedTopHits firstWork = bucket.getAggregations().get("firstWork");
                    // 该用户首次工单ID
                    Long workId = getId(firstWork);
                    CoupUserWork coupUserWork = userWorkMap.get(keyAsNumber.longValue());
                    Long firstWorkIdUseCoupons = coupUserWork.getWorkId();
                    return firstWorkIdUseCoupons.equals(workId);
                })
                .collect(Collectors.toList());
        coupBatchFact.setNewCount(newCountBucket.size());
        List<Long> newCountUserId = newCountBucket.stream().map(Terms.Bucket::getKeyAsNumber).map(Number::longValue).collect(Collectors.toList());
        List<Long> newCountWorkId = newCountBucket.stream().map(bucket -> {
            ParsedTopHits firstWork = bucket.getAggregations().get("firstWork");
            return getId(firstWork);
        }).collect(Collectors.toList());
        // 求用户复购单量
        // List<Long> userRepurchaseOrder = userWorkMaps.stream()
        //         // 减去非用户新单量用户的工单
        //         .filter(coupUserWork -> newCountUserId.contains(coupUserWork.getUserId()))
        //         // 减去用户新单量里的工单
        //         .filter(coupUserWork -> !newCountWorkId.contains(coupUserWork.getWorkId()))
        //         .map(CoupUserWork::getWorkId)
        //         .collect(Collectors.toList());
        int i = userWorkMaps.size() - newCountBucket.size();
        coupBatchFact.setRepurchaseCount(i);
        ParsedFilter amountSumFilter = aggregations.get("amountSumFilter");
        Aggregations amountSumFilterAggregations = amountSumFilter.getAggregations();
        ParsedFilter internalFilter = ((ParsedNested) amountSumFilterAggregations.get("discountList")).getAggregations().get("filterCoupon");
        double discountAmountSum = ((ParsedSum)internalFilter.getAggregations().get("discountAmountSum")).getValue();
        double originalAmountSum = ((ParsedSum) amountSumFilterAggregations.get("originalAmountSum")).getValue();
        coupBatchFact.setDiscountAmount((long) discountAmountSum);
        coupBatchFact.setOriginalAmount((long) originalAmountSum);
        return coupBatchFact;
    }

    private Long getId(ParsedTopHits firstWork) {
        Object id = firstWork.getHits().getHits()[0].getSourceAsMap().get("id");
        if (id instanceof Integer) {
            return ((Integer) id).longValue();
        }
        return (Long) id;
    }

}
